package com.sdses.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author codejiwei
 */
public class FlinkSQL_Kafka2Kafka {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(10000);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                " `jhmeter_id` VARCHAR(64) NULL COMMENT '居建热表设备表ID',\n" +
                " `real_time` STRING NOT NULL COMMENT '设备上传时间',\n" +
                " `save_time` STRING NOT NULL COMMENT '入库保存时间',\n" +
                " `task_time` STRING NULL COMMENT '定时时间',\n" +
                " `id` BIGINT NULL COMMENT '主键',\n" +
                " `pd_id` VARCHAR(64) NULL COMMENT '原始数据表ID',\n" +
                " `mpno` VARCHAR(64) NULL COMMENT '设备上传计量点',\n" +
                " `equip_no` VARCHAR(64) NULL COMMENT '设备上传表号',\n" +
                " `equip_type_code` VARCHAR(64) NULL COMMENT '设备上传表类型',\n" +
                " `jhmeter_ljrl` DECIMAL(18, 2) NULL COMMENT '热表累计热量',\n" +
                " `jhmeter_ljll` DECIMAL(18, 2) NULL COMMENT '热表累计流量',\n" +
                " `jhmeter_ssrl` DECIMAL(18, 2) NULL COMMENT '热表瞬时热量',\n" +
                " `jhmeter_ssll` DECIMAL(18, 2) NULL COMMENT '热表瞬时流量',\n" +
                " `jhmeter_gswd` DECIMAL(18, 2) NULL COMMENT '热表供水温度',\n" +
                " `jhmeter_hswd` DECIMAL(18, 2) NULL COMMENT '热表回水温度',\n" +
                " `jhmeter_wc` DECIMAL(18, 2) NULL COMMENT '热表供回温差',\n" +
                " `jhmeter_yxsj` DECIMAL(18, 2) NULL COMMENT '热表运行时间',\n" +
                " `jhmeter_last_ljrl` DECIMAL(18, 2) NULL COMMENT '上次累计热量',\n" +
                " `jhmeter_cyrl` DECIMAL(18, 2) NULL COMMENT '热表次用热量',\n" +
                " `jhmeter_ryrl` DECIMAL(18, 2) NULL COMMENT '热表日用热量',\n" +
                " `jhmeter_last_ljll` DECIMAL(18, 2) NULL COMMENT '上次累计流量',\n" +
                " `jhmeter_cyll` DECIMAL(18, 2) NULL COMMENT '热表次用流量',\n" +
                " `jhmeter_ryll` DECIMAL(18, 2) NULL COMMENT '热表日用流量',\n" +
                " `jhmeter_last_yxsj` DECIMAL(18, 2) NULL COMMENT '上次运行时间',\n" +
                " `jhmeter_cyxsj` DECIMAL(18, 2) NULL COMMENT '热表次运行时间',\n" +
                " `jhmeter_ryxsj` DECIMAL(18, 2) NULL COMMENT '热表日运行时间',\n" +
                " `alarm_status` VARCHAR(64) NULL COMMENT '报警代码',\n" +
                " `error_detail` VARCHAR(6000) NULL COMMENT '报警内容',\n" +
                " `extend_field` VARCHAR(6000) NULL COMMENT '扩展字段',\n" +
                " `tz_jhmeter_ssrl` DECIMAL(18, 2) NULL COMMENT '调整后的热表瞬时热量',\n" +
                " `tz_jhmeter_ssll` DECIMAL(18, 2) NULL COMMENT '调整后的热表瞬时流量',\n" +
                " `tz_jhmeter_gswd` DECIMAL(18, 2) NULL COMMENT '调整后的热表进水温度',\n" +
                " `tz_jhmeter_hswd` DECIMAL(18, 2) NULL COMMENT '调整后的热表回水温度',\n" +
                " `tz_jhmeter_wc` DECIMAL(18, 2) NULL COMMENT '调整后的热表供回温差',\n" +
                " `instantaneous_heat` DECIMAL(18, 2) NULL COMMENT '瞬时热量偏差',\n" +
                " `instantaneous_flow` DECIMAL(18, 2) NULL COMMENT '瞬时流量偏差',\n" +
                " `inlet_water_temperature` DECIMAL(18, 2) NULL COMMENT '进水温度偏差',\n" +
                " `water_circulation_temperature` DECIMAL(18, 2) NULL COMMENT '回水温度偏差',\n" +
                " `temperature_difference` DECIMAL(18, 2) NULL COMMENT '温度差偏差',\n" +
                " `status` INT NULL COMMENT '1:成功  2:表计报警  4:表计通讯失败 8:采集模块忙  16:EB90解析错误 38:数据清洗 256:数据报警',\n" +
                " `abnormal_data` VARCHAR(2000) NULL COMMENT '异常具体数据'\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 't_data_wdy_real',\n" +   //
                "  'properties.bootstrap.servers' = '172.21.32.233:39092,172.21.32.234:39092,172.21.32.235:39092',\n" +
                "  'properties.group.id' = 'test_2024101211',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +  //'earliest-offset'，'latest-offset'，'group-offsets'，'timestamp' 和 'specific-offsets'
                "  'json.ignore-parse-errors' = 'true',\n" +
                "  'format' = 'json'\n" +
                ")");

//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");

        //KafkaSink
        tableEnv.executeSql("CREATE TABLE sinkTable WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'jw_test',\n" +
                "  'properties.bootstrap.servers' = '172.21.32.233:39092,172.21.32.234:39092,172.21.32.235:39092',\n" +
                "  'format' = 'json'\n" +               //json格式序列化:{"id":1,"sid":"11baac5a","ts":"2023-06-12 15:09:22.849","flag":true,"vc":31}
                ") LIKE sourceTable (EXCLUDING ALL)");

        //DorisSink
//        tableEnv.executeSql("CREATE TABLE sinkTable WITH (\n" +
//                "'connector' = 'doris',\n" +
//                "'fenodes' = '192.168.102.114:38030',\n" +
//                "'table.identifier' = 'db_datacube_doris_test.ods_sc01_rlwlyyxt_t_data_jhmeter_real',\n" +
//                "'username' = 'root',\n" +
//                "'password' = 'Tk6X_U8flz',\n" +
//                "'sink.properties.format' = 'json',\n" +
//                "'sink.properties.read_json_by_line' = 'true',\n" +
//                "'sink.enable-delete' = 'true',\n" +
//                "'sink.label-prefix' = 'doris_label123'\n" +
//                ") LIKE sourceTable (EXCLUDING ALL)");
        //Insert
        tableEnv.executeSql("insert into sinkTable select * from sourceTable");
    }
}